package com.xixi.canal.client.starter.loader;

import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.xixi.canal.client.DataWriter;
import com.xixi.canal.client.starter.config.CanalConfig;
import com.xixi.canal.client.starter.config.InstanceConfig;
import com.xixi.canal.client.starter.config.SpringContext;
import com.xixi.canal.client.support.ExtensionLoader;
import com.xixi.canal.client.support.WriterConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class CanalClientLoader {

    private static final Logger logger = LoggerFactory.getLogger(CanalClientLoader.class);

    private final CanalConfig canalConfig;

    private final CanalServerWithEmbedded embeddedServer;

    private Map<String, CanalClientWorker> workers = new HashMap<>();

    private ExtensionLoader<DataWriter> loader;

    private ConcurrentMap<String, InstanceConfig> instancesConfig;


    public CanalClientLoader(CanalConfig canalConfig,
                             CanalServerWithEmbedded embeddedServer, ConcurrentMap<String, InstanceConfig> instancesConfig) {

        this.canalConfig = canalConfig;
        this.embeddedServer = embeddedServer;
        this.instancesConfig = instancesConfig;
    }

    public void init() {

        loader = ExtensionLoader.getExtensionLoader(DataWriter.class);

        for (Map.Entry<String, InstanceConfig> entry : instancesConfig.entrySet()) {

            String destination = entry.getKey();
            InstanceConfig instanceConfig = entry.getValue();

            // load data writer
            List<DataWriter> writers = new ArrayList<>();

            for (WriterConfig writerConfig : instanceConfig.getWriters()) {
                writerConfig.setConfDir(canalConfig.getConfDir());
                loadWriter(destination, writerConfig, writers);
            }

            if (writers.isEmpty()) {
                logger.error("No client writer for instance:{}.", destination);
                continue;
            }

            CanalClientWorker worker = new CanalClientWorker(instanceConfig,
                    destination, writers, embeddedServer);

            workers.put(destination, worker);
            worker.start();
            logger.info("Start client for canal instance: {} succeed", destination);
        }
    }

    private void loadWriter(String destination, WriterConfig writerConfig, List<DataWriter> writers) {

        try {
            DataWriter writer = loader.getExtension(writerConfig.getName(), writerConfig.getKey());

            ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();

            Thread.currentThread().setContextClassLoader(writer.getClass().getClassLoader());

            Properties envProperties = SpringContext.getEnvProperties();

            writer.init(destination, writerConfig, envProperties);
            Thread.currentThread().setContextClassLoader(contextClassLoader);
            writers.add(writer);
            logger.info("Load canal client writer:{} success.", writerConfig.getName());
        } catch (Exception e) {
            logger.error("Load canal client writer:{} failed.{}", writerConfig.getName(), e);
        }
    }

    public void destroy() {

        if (!workers.isEmpty()) {
            ExecutorService executorService = Executors.newFixedThreadPool(workers.size());
            for (CanalClientWorker worker : workers.values()) {
                executorService.execute(worker::stop);
            }
            executorService.shutdown();
            try {
                while (!executorService.awaitTermination(1, TimeUnit.SECONDS)) {
                    // ignore
                }
            } catch (InterruptedException e) {
                // ignore
            }
        }

        if (null != instancesConfig && !instancesConfig.isEmpty()) {
            instancesConfig.clear();
            instancesConfig = null;
        }
        logger.info("All canal writers destroyed");

    }
}
